package com.fine.spark.connector.base.worker.process

import com.fine.spark.connector.base.beans.BaseConfig
import com.fine.spark.connector.base.beans.process.ClassProcessConfig
import com.fine.spark.connector.base.utils.ReflectUtils
import com.fine.spark.connector.base.worker.{BaseWorker, ClassBaseWorker}
import org.apache.spark.sql.SparkSession

/**
 * @author jayce
 * @date 2021/11/21 3:00 PM
 * @version 1.0
 */
class ClassProcessWorker extends BaseWorker {
  /**
   * 定义处理过程的方法
   *
   * @param bean
   * @param ss
   */
  override def process(config: BaseConfig)(implicit ss: SparkSession): Unit = {
    val processConfig: ClassProcessConfig = config.asInstanceOf[ClassProcessConfig]
    ReflectUtils.apply.getInstance[ClassBaseWorker](processConfig.clazz).main(processConfig)
    if (ss.catalog.tableExists(processConfig.viewName)) {
      afterProcess(processConfig)
    }
  }
}


object ClassProcessWorker {
  def apply: ClassProcessWorker = new ClassProcessWorker()
}